package mr

import (
	"bufio"
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"sort"
	"strings"
	"time"
)

//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

type KeyValues struct {
	Key    string
	Values []string
}

// for sorting by key.
type SortKey []KeyValues

// for sorting by key.
func (a SortKey) Len() int           { return len(a) }
func (a SortKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a SortKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

//
// main/mrworker.go calls this function.
//

func doMap(reply *TaskReply, mapf func(string, string) []KeyValue) {
	intermediate := []KeyValue{}
	for _, filename := range reply.InputFilename {
		file, err := os.Open(filename)
		if err != nil {
			log.Fatalf("cannot open %v", filename)
		}
		content, err := ioutil.ReadAll(file)
		if err != nil {
			log.Fatalf("cannot read %v", filename)
		}
		file.Close()
		kva := mapf(filename, string(content))
		intermediate = append(intermediate, kva...)
	}

	ofilename := reply.OutputFilename
	ofile, _ := os.Create(ofilename)

	for _, kv := range intermediate {
		i := ihash(kv.Key) % reply.NReduce
		if i == reply.FileIndex {
			fmt.Fprintf(ofile, "%v %v\n", kv.Key, kv.Value)
		}
	}
	ofile.Close()
}

func doReduce(reply *TaskReply, reducef func(string, []string) string) {
	intermediate := make(map[string][]string)
	for _, filename := range reply.InputFilename {
		file, err := os.Open(filename)
		if err != nil {
			log.Fatalf("cannot open %v", filename)
		}
		scanner := bufio.NewScanner(file)
		for scanner.Scan() {
			kva := strings.Split(scanner.Text(), " ")

			intermediate[kva[0]] = append(intermediate[kva[0]], kva[1])
		}
		if scanner.Err() != nil {
			log.Fatalln(scanner.Err())
		}
		file.Close()
	}

	inters := make([]KeyValues, 0)
	for k, v := range intermediate {
		kvs := KeyValues{Key: k, Values: v}
		inters = append(inters, kvs)
	}
	sort.Sort(SortKey(inters))

	oname := reply.OutputFilename
	ofile, _ := os.Create(oname)
	for i := 0; i < len(inters); i++ {
		output := reducef(inters[i].Key, inters[i].Values)
		fmt.Fprintf(ofile, "%v %v\n", inters[i].Key, output)
	}

	ofile.Close()
}

func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.

	// uncomment to send the Example RPC to the master.
	//CallExample()
	for {
		args := &TaskArgs{}
		reply := &TaskReply{}
		if requestCall(args, reply) == false {
			return
		}
		if reply.Phase == MapPhase {
			doMap(reply, mapf)
			args.Phase = reply.Phase
			args.FileIndex = reply.FileIndex
			args.Id = reply.Id
			// fmt.Printf("worker task id: %v, %v\n", args.Id, reply.Id)

			reportCall(args, reply)
		} else if reply.Phase == ReducePhase {
			// fmt.Printf("Worker: do Reduce, %v\n", reply)
			doReduce(reply, reducef)
			args.Phase = reply.Phase
			args.FileIndex = reply.FileIndex
			args.Id = reply.Id
			reportCall(args, reply)
		} else if reply.Phase == WaitPhase {
			// fmt.Println("Worker: Recevice Sleep  ... ")
			time.Sleep(time.Duration(1) * time.Second)
		} else {
			return
		}
	}
}

func requestCall(args *TaskArgs, reply *TaskReply) bool {
	return call("Master.RequestTask", args, reply)
}

func reportCall(args *TaskArgs, reply *TaskReply) bool {
	return call("Master.ReportTask", args, reply)
}

//
// example function to show how to make an RPC call to the master.
//
// the RPC argument and reply types are defined in rpc.go.
//
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	call("Master.Example", &args, &reply)

	// reply.Y should be 100.
	fmt.Printf("reply.Y %v\n", reply.Y)
}

//
// send an RPC request to the master, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := masterSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
